package com.we.flink.function;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSON;
import com.we.flink.bean.FlinkCheckpointInfo;

import java.sql.Timestamp;

/**
 * @author ：Jason
 * @date ：Created in 2023/5/31 6:07 PM
 * @description：
 * @modified By：
 * @version: 1.0
 */
public class FlinkCheckpointKeyedProcessFunction
        extends KeyedProcessFunction<String, FlinkCheckpointInfo, String> {
    private static final long TIMER_TIME = 360 * 1000L;
    private transient ValueState<FlinkCheckpointInfo> valueState;

    @Override
    public void open(Configuration parameters) throws Exception {
        valueState =
                getRuntimeContext()
                        .getState(
                                new ValueStateDescriptor<>(
                                        "BorrowInfo", FlinkCheckpointInfo.class));
    }

    @Override
    public void processElement(
            FlinkCheckpointInfo value,
            KeyedProcessFunction<String, FlinkCheckpointInfo, String>.Context ctx,
            Collector<String> out)
            throws Exception {
        valueState.update(value);
        out.collect(JSON.toJSONString(value));
        // update_time + 6 min 注册 timer
        ctx.timerService()
                .registerProcessingTimeTimer(
                        Timestamp.valueOf(value.getUpdateTime()).getTime() + TIMER_TIME);
    }

    @Override
    public void onTimer(
            long timestamp,
            KeyedProcessFunction<String, FlinkCheckpointInfo, String>.OnTimerContext ctx,
            Collector<String> out)
            throws Exception {
        final FlinkCheckpointInfo lastFlinkCheckpointState = valueState.value();
        final long updateTime =
                Timestamp.valueOf(lastFlinkCheckpointState.getUpdateTime()).getTime();
        // 说明 6 分钟没有更新数据了,就把任务状态设置为 FINISHED 并且删除掉触发器
        if (System.currentTimeMillis() - updateTime > TIMER_TIME) {
            lastFlinkCheckpointState.setJobState("FINISHED");
            out.collect(JSON.toJSONString(lastFlinkCheckpointState));
            valueState.clear();
            ctx.timerService().deleteProcessingTimeTimer(updateTime + TIMER_TIME);
        }
    }
}
